#include "my_aio.h"

static my_timespec_t my_cond_wait_time = {
    .tv_sec = 0,
    .tv_nsec = 100000000,       // 100ms.
};

static inline void
my_aio_loop_executer(my_aio_loop_data_t *loop_data)
{
    my_queue_t              *q = NULL;
    my_dispatch_event_t     *dispatch_event;
    my_aio_request_t        *aio_req;

    my_mutex_lock(&loop_data->mutex);
    while (my_queue_empty(&loop_data->extern_queue) &&
        !my_atomic_int64_load(&loop_data->stop)) {
        my_cond_time_wait(&loop_data->cond, &loop_data->mutex, &my_cond_wait_time);
    }

    if (!my_queue_empty(&loop_data->extern_queue)) {
        q = my_queue_last(&loop_data->extern_queue);
        my_queue_remove(q);
    }

    my_mutex_unlock(&loop_data->mutex);

    if (!q) {
        return; // aio loop stop
    }

    dispatch_event = my_queue_data(q, my_dispatch_event_t, queue);
    aio_req = (my_aio_request_t *)dispatch_event->save_data;

    switch (aio_req->op) {
    case MY_AIO_READ:
        aio_req->res =
            my_syscall_pread(aio_req->fd, aio_req->buf, aio_req->nbytes, aio_req->offset);
        break;
    case MY_AIO_WRITE:
        aio_req->res =
            my_syscall_pwrite(aio_req->fd, aio_req->buf, aio_req->nbytes, aio_req->offset);
        break;
    default:
        break;
    }

    my_atomic_int64_add_fetch(&loop_data->current_task, -1);

    aio_req->down_time = my_usec_2_msec(get_cache_time());

    my_event_loop_post_event(dispatch_event->setup_thread, dispatch_event);
}

static inline void
my_aio_loop_runner(my_thread_base_t *thread_base)
{
    my_aio_loop_data_t *loop_data = thread_base->data;

    for ( ;; ) {
        if (my_atomic_int64_load(&loop_data->stop) &&
            my_atomic_int64_load(&loop_data->current_task) == 0) {
            break;
        }

        my_aio_loop_executer(loop_data);
    }
}

void
my_aio_manager_start(my_aio_manager_t *aio_manager,
    int n_read_threads, int n_write_threads,
    int64_t read_task_limit, int64_t write_task_limit)
{
    if (n_read_threads <= 0 || n_write_threads <= 0) {
        MY_LOG_FATAL("n_read_threads or n_write_threads must be greate than 0");
    }

    bzero(aio_manager, sizeof(my_aio_manager_t));
    aio_manager->n_read_threads = n_read_threads;
    aio_manager->n_write_threads = n_write_threads;

    my_aio_loop_data_init(&aio_manager->read_loop_data, read_task_limit);
    my_aio_loop_data_init(&aio_manager->write_loop_data, write_task_limit);

    aio_manager->read_threads =
        my_thread_calloc((int64_t)sizeof(my_thread_base_t) * n_read_threads);
    aio_manager->write_threads =
        my_thread_calloc((int64_t)sizeof(my_thread_base_t) * n_write_threads);

    for (int i = 0; i < n_read_threads; i++) {
        my_thread_base_start(aio_manager->read_threads + i,
            my_aio_loop_runner,
            &aio_manager->read_loop_data,
            0, 0);
    }

    for (int i = 0; i < n_write_threads; i++) {
        my_thread_base_start(aio_manager->write_threads + i,
            my_aio_loop_runner,
            &aio_manager->write_loop_data,
            0, 0);
    }
}

void
my_aio_manager_stop(my_aio_manager_t *aio_manager)
{
    my_atomic_int64_swap(&aio_manager->read_loop_data.stop, 1);
    my_atomic_int64_swap(&aio_manager->write_loop_data.stop, 1);

    for (int i = 0; i < aio_manager->n_read_threads; i++) {
        my_thread_base_join(aio_manager->read_threads + i, NULL);
    }

    for (int i = 0; i < aio_manager->n_write_threads; i++) {
        my_thread_base_join(aio_manager->write_threads + i, NULL);
    }

    my_thread_free(aio_manager->read_threads);
    my_thread_free(aio_manager->write_threads);
}

my_bool_t
my_aio_post_event(my_aio_manager_t *aio_manager, my_aio_request_t *req)
{
    my_aio_loop_data_t *loop_data = &aio_manager->read_loop_data;

    if (req->op == MY_AIO_WRITE || req->op == MY_AIO_WRITE_PING) {
        loop_data = &aio_manager->write_loop_data;
    }

    // QUEUE BUSY.
    if (loop_data->tasks_limit > 0 &&
        my_atomic_int64_load(&loop_data->current_task) > loop_data->tasks_limit) {
        return MY_FALSE;
    }

    req->dispatch_event.setup_thread = current_thread_base();
    req->setup_time = my_usec_2_msec(get_cache_time());

    my_mutex_lock(&loop_data->mutex);
    my_queue_insert_head(&loop_data->extern_queue, &req->dispatch_event.queue);
    my_cond_signal(&loop_data->cond);
    my_atomic_int64_add_fetch(&loop_data->current_task, 1);
    my_mutex_unlock(&loop_data->mutex);

    return MY_TRUE;
}